package io.kimmking.common.rocketmq.config.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.Closeable;

/**
 * 根据获取到的MQ信息，启动对应的消费者
 */
@EnableConfigurationProperties(RocketMQProperties.class)
public class StartConsumer implements Closeable {

    @Resource
    private RocketMQProperties rocketMQProperties;

    private DefaultMQPushConsumer consumer;

    private final MessageListenerConcurrently messageListener;

    private final String topic;

    private final String consumer_group;

    public StartConsumer(MessageListenerConcurrently messageListener, String topic, String consumer_group){
        this.consumer_group = consumer_group;
        this.messageListener = messageListener;
        this.topic = topic;
    }

    @PostConstruct
    public void init() throws Exception{
        consumer = new DefaultMQPushConsumer(consumer_group);
        //服务地址
        consumer.setNamesrvAddr(rocketMQProperties.getNameServer());
        //订阅主题
        if("TAG".equals(rocketMQProperties.getConsumer().getSelectorType())) {//如果是根据tag过滤消息
            consumer.subscribe(topic, rocketMQProperties.getConsumer().getSelectorExpression());
        }else if("SQL92".equals(rocketMQProperties.getConsumer().getSelectorType())) {//如果是根据sql过滤消息
            consumer.subscribe(topic, MessageSelector.bySql(rocketMQProperties.getConsumer().getSelectorExpression()));
        }
        //消息模式
        consumer.setMessageModel("CLUSTERING".
                equals(rocketMQProperties.getConsumer().getMessageModel())
                ? MessageModel.CLUSTERING : MessageModel.BROADCASTING);
        //监听器
        consumer.registerMessageListener(messageListener);
        consumer.start();
    }

    @Override
    public void close() {
        if(consumer != null)
            consumer.shutdown();
    }
}
